--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
--------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------
Commit 6f92ad865e4c1a2943fe54fb119a9ac8d48d1318
Parents : 29d9178
Author : Ivan <ivan@quad4.io>
Signature : Invalid signer <e46112d44649266d71fe2193e00a4710>, author is <ivan@quad4.io>
Date : 2026-07-09T14:04:46-05:00
feat(telephone): update audio profile management and Codec2 readiness; implement profile migration, add Codec2 status endpoint, and improve audio level reporting
Changes
15 files changed, 458 insertions(+), 25 deletions(-)
Diff
diff --git a/android/app/src/main/java/com/meshchatx/TelephoneNativeAudioSession.java b/android/app/src/main/java/com/meshchatx/TelephoneNativeAudioSession.java
index 838810b8..db28fbf2 100644
--- a/android/app/src/main/java/com/meshchatx/TelephoneNativeAudioSession.java
+++ b/android/app/src/main/java/com/meshchatx/TelephoneNativeAudioSession.java
@@ -321,6 +321,7 @@ public final class TelephoneNativeAudioSession {
private void drainMicrophone() {
byte[] buf = new byte[4096];
+ long lastLevelsMs = 0L;
while (running.get() && webSocket != null) {
AudioRecord ar;
WebSocket w;
@@ -343,12 +344,20 @@ public final class TelephoneNativeAudioSession {
} catch (Exception e) {
break;
}
+ long now = System.currentTimeMillis();
+ if (now - lastLevelsMs >= 200L) {
+ lastLevelsMs = now;
+ float tx = pcmPeakLevel(buf, n);
+ postLevels(tx, lastRxLevel);
+ }
} else if (n < 0) {
break;
}
}
}
+ private volatile float lastRxLevel = 0f;
+
private void playPcm(ByteString bytes) {
if (bytes == null || bytes.size() == 0) {
return;
@@ -361,6 +370,7 @@ public final class TelephoneNativeAudioSession {
return;
}
byte[] raw = bytes.toByteArray();
+ lastRxLevel = pcmPeakLevel(raw, raw.length);
int off = 0;
int left = raw.length;
while (left > 0) {
@@ -391,6 +401,50 @@ public final class TelephoneNativeAudioSession {
}
}
+ private static float pcmPeakLevel(byte[] pcm, int length) {
+ if (pcm == null || length < 2) {
+ return 0f;
+ }
+ int peak = 0;
+ int end = length - (length % 2);
+ for (int i = 0; i < end; i += 2) {
+ int sample = (pcm[i] & 0xff) | (pcm[i + 1] << 8);
+ if (sample > 32767) {
+ sample -= 65536;
+ }
+ int abs = sample < 0 ? -sample : sample;
+ if (abs > peak) {
+ peak = abs;
+ }
+ }
+ return Math.min(1f, peak / 32767f);
+ }
+
+ private void postLevels(float tx, float rx) {
+ activity.runOnUiThread(() -> dispatchLevels(tx, rx));
+ }
+
+ private void dispatchLevels(float tx, float rx) {
+ try {
+ WebView wv = activity.getWebViewForNativeBridge();
+ if (wv == null) {
+ return;
+ }
+ JSONObject o = new JSONObject();
+ o.put("type", "meshchatx-native-telephone-audio");
+ o.put("kind", "levels");
+ o.put("tx_level", tx);
+ o.put("rx_level", rx);
+ String j = o.toString();
+ wv.evaluateJavascript(
+ "try{var d=" + j + ";"
+ + "window.dispatchEvent(new CustomEvent('meshchatx-native-telephone-audio',{detail:d}));}catch(e){}",
+ null
+ );
+ } catch (Exception ignored) {
+ }
+ }
+
private void postDispatch(String a, @Nullable String b, @Nullable String c) {
activity.runOnUiThread(() -> dispatchToPage(a, b, c));
}
diff --git a/android/app/src/main/python/meshchat_wrapper.py b/android/app/src/main/python/meshchat_wrapper.py
index 421b2f39..ee9fef11 100644
--- a/android/app/src/main/python/meshchat_wrapper.py
+++ b/android/app/src/main/python/meshchat_wrapper.py
@@ -122,9 +122,14 @@ def start_server(port=8000, app_files_dir=None):
asyncio_signal_patch = _patch_asyncio_signal_handlers_for_android()
aiohttp_run_app_patch = _patch_aiohttp_run_app_for_android()
try:
- from meshchatx.android_codec2 import ensure_codec2_native_library
+ from meshchatx.android_codec2 import ensure_codec2_native_library, probe_pycodec2
ensure_codec2_native_library()
+ ok, err = probe_pycodec2()
+ if ok:
+ print("meshchat_wrapper: Codec2/pycodec2 ready")
+ else:
+ print(f"meshchat_wrapper: Codec2/pycodec2 unavailable: {err}")
except Exception as codec2_exc:
print(f"meshchat_wrapper: Codec2 preload skipped: {codec2_exc}")
from meshchatx.meshchat import ReticulumMeshChat, main
diff --git a/meshchatx/android_codec2.py b/meshchatx/android_codec2.py
index 4ba3a40e..debc3897 100644
--- a/meshchatx/android_codec2.py
+++ b/meshchatx/android_codec2.py
@@ -89,6 +89,20 @@ def ensure_codec2_native_library() -> bool:
return False
+def probe_pycodec2() -> tuple[bool, str | None]:
+ """Import pycodec2 after preload and report whether Codec2 works."""
+ if _is_chaquopy_android() and not ensure_codec2_native_library():
+ return False, codec2_preload_error()
+ try:
+ import pycodec2
+
+ c2 = pycodec2.Codec2(1600)
+ _ = c2.samples_per_frame()
+ return True, None
+ except Exception as exc:
+ return False, str(exc)
+
+
def codec2_preload_error() -> str | None:
"""Return the last preload failure message, if any."""
return _codec2_preload_error
diff --git a/meshchatx/meshchat.py b/meshchatx/meshchat.py
index ff3d83da..afc4ba5e 100644
--- a/meshchatx/meshchat.py
+++ b/meshchatx/meshchat.py
@@ -9767,16 +9767,37 @@ class ReticulumMeshChat:
status=400,
)
- await asyncio.to_thread(
- self.telephone_manager.telephone.switch_profile,
+ resolved = await asyncio.to_thread(
+ self.telephone_manager.apply_preferred_profile,
int(profile_id),
)
+ self.config.telephone_audio_profile_id.set(resolved)
return web.json_response(
- {"message": f"Switched to profile {profile_id}"},
+ {
+ "message": f"Switched to profile {resolved}",
+ "profile_id": resolved,
+ },
)
except Exception as e:
return web.json_response({"message": str(e)}, status=500)
+ # Codec2 / audio backend readiness (Android packaging + LXST)
+ @routes.get("/api/v1/telephone/codec2/status")
+ async def telephone_codec2_status(request):
+ from meshchatx import android_codec2
+
+ available = await asyncio.to_thread(
+ self.telephone_manager.codec2_available,
+ )
+ return web.json_response(
+ {
+ "codec2_available": available,
+ "preload_error": android_codec2.codec2_preload_error(),
+ "preferred_profile_id": self.telephone_manager.preferred_profile_id,
+ "resolved_profile_id": self.telephone_manager.resolve_audio_profile_id(),
+ },
+ )
+
# initiate a telephone call
# initiate outgoing telephone call
@routes.get("/api/v1/telephone/call/{identity_hash}")
@@ -16945,9 +16966,9 @@ class ReticulumMeshChat:
if profile_id is None:
profile_id = self.config.telephone_audio_profile_id.get()
self.config.telephone_audio_profile_id.set(profile_id)
- if self.telephone_manager and self.telephone_manager.telephone:
+ if self.telephone_manager:
await asyncio.to_thread(
- self.telephone_manager.telephone.switch_profile,
+ self.telephone_manager.apply_preferred_profile,
profile_id,
)
diff --git a/meshchatx/src/backend/config_manager.py b/meshchatx/src/backend/config_manager.py
index 74214bbf..97e3d0f7 100644
--- a/meshchatx/src/backend/config_manager.py
+++ b/meshchatx/src/backend/config_manager.py
@@ -254,7 +254,7 @@ class ConfigManager:
self.telephone_audio_profile_id = self.IntConfig(
self,
"telephone_audio_profile_id",
- 2, # Default to Voice (profile 2)
+ 64, # LXST Profiles.DEFAULT_PROFILE (Medium Quality / Opus)
)
self.telephone_web_audio_enabled = self.BoolConfig(
self,
@@ -589,6 +589,7 @@ class ConfigManager:
self._migrate_legacy_announce_limit_keys()
self._migrate_translator_from_legacy()
+ self._migrate_invalid_telephone_audio_profile()
def get(self, key: str, default_value=None) -> str | None:
return self.db.config.get(key, default_value)
@@ -596,6 +597,20 @@ class ConfigManager:
def set(self, key: str, value: str | None):
self.db.config.set(key, value)
+ def _migrate_invalid_telephone_audio_profile(self):
+ """Reset stale profile ids (e.g. legacy default 2) to LXST DEFAULT_PROFILE."""
+ raw = self.db.config.get("telephone_audio_profile_id", default=None)
+ if raw is None:
+ return
+ try:
+ pid = int(raw)
+ except (TypeError, ValueError):
+ self.telephone_audio_profile_id.set(64)
+ return
+ valid = {16, 32, 48, 64, 80, 96, 112, 128}
+ if pid not in valid:
+ self.telephone_audio_profile_id.set(64)
+
def _migrate_translator_from_legacy(self):
old = self.db.config.get("translator_enabled", default=None)
a = self.db.config.get("translator_argos_enabled", default=None)
diff --git a/meshchatx/src/backend/telephone_manager.py b/meshchatx/src/backend/telephone_manager.py
index 0b23c46b..54ebeb7a 100644
--- a/meshchatx/src/backend/telephone_manager.py
+++ b/meshchatx/src/backend/telephone_manager.py
@@ -89,11 +89,68 @@ class TelephoneManager:
self._path_retry_interval_s = 1.5
self._status_poll_interval_s = 0.1
self.is_voicemail_session_active = False
+ self.preferred_profile_id = None
@property
def is_recording(self):
return False
+ @staticmethod
+ def codec2_available() -> bool:
+ """Return whether LXST can construct Codec2 codecs (pycodec2 + libcodec2)."""
+ try:
+ from LXST.Codecs import Codec2
+
+ if Codec2 is None:
+ return False
+ # Touch a mode constant and construct once to catch dlopen failures early.
+ _ = Codec2.CODEC2_1600
+ Codec2(mode=Codec2.CODEC2_1600)
+ return True
+ except Exception:
+ return False
+
+ def resolve_audio_profile_id(self, profile_id=None):
+ """Return a valid LXST profile id, falling back when Codec2 is unavailable."""
+ from LXST.Primitives.Telephony import Profiles
+
+ available = set(Profiles.available_profiles())
+ pid = profile_id
+ if pid is None and self.config_manager:
+ with contextlib.suppress(Exception):
+ pid = self.config_manager.telephone_audio_profile_id.get()
+ try:
+ pid = int(pid) if pid is not None else None
+ except (TypeError, ValueError):
+ pid = None
+ if pid not in available:
+ pid = Profiles.DEFAULT_PROFILE
+
+ codec2_profiles = {
+ Profiles.BANDWIDTH_ULTRA_LOW,
+ Profiles.BANDWIDTH_VERY_LOW,
+ Profiles.BANDWIDTH_LOW,
+ }
+ if pid in codec2_profiles and not self.codec2_available():
+ RNS.log(
+ "TelephoneManager: Codec2 unavailable, falling back to default Opus profile",
+ RNS.LOG_WARNING,
+ )
+ pid = Profiles.DEFAULT_PROFILE
+ return pid
+
+ def apply_preferred_profile(self, profile_id=None):
+ """Store preferred profile and apply it if a call is already established."""
+ self.preferred_profile_id = self.resolve_audio_profile_id(profile_id)
+ if (
+ self.telephone
+ and self.telephone.active_call
+ and self.telephone.call_status == 6
+ ):
+ with contextlib.suppress(Exception):
+ self.telephone.switch_profile(self.preferred_profile_id)
+ return self.preferred_profile_id
+
def init_telephone(self):
if self.telephone is not None:
return
@@ -106,10 +163,9 @@ class TelephoneManager:
# Increase connection timeout for slower networks
self.telephone.set_connect_timeout(30)
- # Set initial profile from config
- if self.config_manager:
- profile_id = self.config_manager.telephone_audio_profile_id.get()
- self.telephone.switch_profile(profile_id)
+ # LXST switch_profile is a no-op without an established call. Remember the
+ # preferred profile and pass it into telephone.call() on outbound dial.
+ self.preferred_profile_id = self.resolve_audio_profile_id()
self.telephone.set_ringing_callback(self.on_telephone_ringing)
self.telephone.set_established_callback(self.on_telephone_call_established)
@@ -385,10 +441,18 @@ class TelephoneManager:
self.call_start_time = time.time()
self.call_is_incoming = False
+ profile_id = self.resolve_audio_profile_id(self.preferred_profile_id)
+ self.preferred_profile_id = profile_id
+
# Use a thread for the blocking LXST call, but monitor status for early exit
- # if established elsewhere or timed out/hung up
+ # if established elsewhere or timed out/hung up. Pass preferred profile so
+ # Codec2/Opus selection actually applies (switch_profile alone is a no-op idle).
call_task = asyncio.create_task(
- asyncio.to_thread(self.telephone.call, destination_identity),
+ asyncio.to_thread(
+ self.telephone.call,
+ destination_identity,
+ profile_id,
+ ),
)
start_wait = time.time()
diff --git a/scripts/build-android-wheels-local.sh b/scripts/build-android-wheels-local.sh
index b22c34d3..e314c7dd 100755
--- a/scripts/build-android-wheels-local.sh
+++ b/scripts/build-android-wheels-local.sh
@@ -750,6 +750,36 @@ with zipfile.ZipFile(src, "r") as zin, zipfile.ZipFile(dst, "w", compression=zip
data = zin.read(item.filename)
if item.filename == "LXST/Codecs/__init__.py":
data = patched_codecs_init.encode("utf-8")
+ elif item.filename == "LXST/Primitives/Telephony.py":
+ text = data.decode("utf-8")
+ old_get_codec = """ @staticmethod
+ def get_codec(profile):
+ if profile == Profiles.BANDWIDTH_ULTRA_LOW: return Codec2(mode=Codec2.CODEC2_700C)
+ elif profile == Profiles.BANDWIDTH_VERY_LOW: return Codec2(mode=Codec2.CODEC2_1600)
+ elif profile == Profiles.BANDWIDTH_LOW: return Codec2(mode=Codec2.CODEC2_3200)
+ elif profile == Profiles.QUALITY_MEDIUM: return Opus(profile=Opus.PROFILE_VOICE_MEDIUM)
+ elif profile == Profiles.QUALITY_HIGH: return Opus(profile=Opus.PROFILE_VOICE_HIGH)
+ elif profile == Profiles.QUALITY_MAX: return Opus(profile=Opus.PROFILE_VOICE_MAX)
+ elif profile == Profiles.LATENCY_LOW: return Opus(profile=Opus.PROFILE_VOICE_MEDIUM)
+ elif profile == Profiles.LATENCY_ULTRA_LOW: return Opus(profile=Opus.PROFILE_VOICE_MEDIUM)
+ else: return Opus(profile=Opus.PROFILE_VOICE_MEDIUM)
+"""
+ new_get_codec = """ @staticmethod
+ def get_codec(profile):
+ if Codec2 is not None:
+ if profile == Profiles.BANDWIDTH_ULTRA_LOW: return Codec2(mode=Codec2.CODEC2_700C)
+ elif profile == Profiles.BANDWIDTH_VERY_LOW: return Codec2(mode=Codec2.CODEC2_1600)
+ elif profile == Profiles.BANDWIDTH_LOW: return Codec2(mode=Codec2.CODEC2_3200)
+ if profile == Profiles.QUALITY_MEDIUM: return Opus(profile=Opus.PROFILE_VOICE_MEDIUM)
+ elif profile == Profiles.QUALITY_HIGH: return Opus(profile=Opus.PROFILE_VOICE_HIGH)
+ elif profile == Profiles.QUALITY_MAX: return Opus(profile=Opus.PROFILE_VOICE_MAX)
+ elif profile == Profiles.LATENCY_LOW: return Opus(profile=Opus.PROFILE_VOICE_MEDIUM)
+ elif profile == Profiles.LATENCY_ULTRA_LOW: return Opus(profile=Opus.PROFILE_VOICE_MEDIUM)
+ else: return Opus(profile=Opus.PROFILE_VOICE_MEDIUM)
+"""
+ if old_get_codec in text:
+ text = text.replace(old_get_codec, new_get_codec)
+ data = text.encode("utf-8")
elif item.filename.endswith(".dist-info/METADATA"):
text = data.decode("utf-8")
text = text.replace("Requires-Dist: numpy>=2.3.4", "Requires-Dist: numpy==${NUMPY_VERSION}")
diff --git a/tests/backend/http_api_response_registry.py b/tests/backend/http_api_response_registry.py
index 7e3499f0..f0b592bc 100644
--- a/tests/backend/http_api_response_registry.py
+++ b/tests/backend/http_api_response_registry.py
@@ -104,6 +104,7 @@ from tests.backend.http_api_response_schemas import (
TELEMETRY_TRUSTED_PEERS_SCHEMA,
TELEPHONE_AUDIO_PROFILES_SCHEMA,
TELEPHONE_CALL_SCHEMA,
+ TELEPHONE_CODEC2_STATUS_SCHEMA,
TELEPHONE_HISTORY_SCHEMA,
TELEPHONE_RECORDINGS_SCHEMA,
TELEPHONE_STATUS_SCHEMA,
@@ -381,6 +382,9 @@ HTTP_JSON_GET_CONTRACTS: tuple[HttpJsonContract, ...] = (
HttpJsonContract(
"GET", "/api/v1/telephone/audio-profiles", TELEPHONE_AUDIO_PROFILES_SCHEMA
),
+ HttpJsonContract(
+ "GET", "/api/v1/telephone/codec2/status", TELEPHONE_CODEC2_STATUS_SCHEMA
+ ),
HttpJsonContract(
"GET",
"/api/v1/telephone/call/{identity_hash}",
diff --git a/tests/backend/http_api_response_schemas.py b/tests/backend/http_api_response_schemas.py
index e7dfaa61..1769e1fb 100644
--- a/tests/backend/http_api_response_schemas.py
+++ b/tests/backend/http_api_response_schemas.py
@@ -702,6 +702,18 @@ TELEPHONE_AUDIO_PROFILES_SCHEMA: dict = {
"additionalProperties": True,
}
+TELEPHONE_CODEC2_STATUS_SCHEMA: dict = {
+ "type": "object",
+ "required": ["codec2_available"],
+ "properties": {
+ "codec2_available": {"type": "boolean"},
+ "preload_error": {"type": ["string", "null"]},
+ "preferred_profile_id": {"type": ["integer", "null"]},
+ "resolved_profile_id": {"type": ["integer", "null"]},
+ },
+ "additionalProperties": True,
+}
+
TELEPHONE_CALL_SCHEMA: dict = {
"type": "object",
"required": ["call"],
diff --git a/tests/backend/test_android_codec2.py b/tests/backend/test_android_codec2.py
index 369b6c7a..de965b6f 100644
--- a/tests/backend/test_android_codec2.py
+++ b/tests/backend/test_android_codec2.py
@@ -36,6 +36,69 @@ def test_ensure_codec2_loads_bundled_library(tmp_path):
cdll.assert_called_with(str(lib))
+def test_probe_pycodec2_reports_failure_when_import_breaks():
+ android_codec2._codec2_preload_done = True
+ android_codec2._codec2_preload_error = None
+ with (
+ patch.object(android_codec2, "_is_chaquopy_android", return_value=False),
+ patch.dict("sys.modules", {"pycodec2": None}),
+ ):
+ # Force ImportError path by making import raise
+ import builtins
+
+ real_import = builtins.__import__
+
+ def fake_import(name, *args, **kwargs):
+ if name == "pycodec2":
+ raise ImportError("no pycodec2")
+ return real_import(name, *args, **kwargs)
+
+ with patch("builtins.__import__", side_effect=fake_import):
+ ok, err = android_codec2.probe_pycodec2()
+ assert ok is False
+ assert err
+
+
+def test_vendor_wheels_bundle_libcodec2_for_all_abis():
+ import zipfile
+
+ repo = Path(__file__).resolve().parents[2]
+ vendor = repo / "android" / "vendor"
+ abis = ("arm64_v8a", "armeabi_v7a", "x86_64")
+ for abi in abis:
+ wheels = sorted(vendor.glob(f"pycodec2-*-android_24_{abi}.whl"))
+ assert wheels, f"missing pycodec2 wheel for {abi}"
+ with zipfile.ZipFile(wheels[-1]) as zin:
+ assert "pycodec2/libcodec2.so" in zin.namelist()
+ assert "pycodec2/pycodec2.so" in zin.namelist()
+ lib_wheels = sorted(vendor.glob(f"chaquopy_libcodec2-*-android_24_{abi}.whl"))
+ assert lib_wheels, f"missing chaquopy_libcodec2 for {abi}"
+ with zipfile.ZipFile(lib_wheels[-1]) as zin:
+ assert "chaquopy/lib/libcodec2.so" in zin.namelist()
+
+
+def test_jni_libs_synced_for_all_abis():
+ repo = Path(__file__).resolve().parents[2]
+ jni = repo / "android" / "app" / "src" / "main" / "jniLibs"
+ for abi in ("arm64-v8a", "armeabi-v7a", "x86_64"):
+ lib = jni / abi / "libcodec2.so"
+ assert lib.is_file(), f"missing jniLibs {lib}"
+ assert lib.stat().st_size > 100_000
+
+
+def test_android_lxst_wheel_get_codec_guards_missing_codec2():
+ import zipfile
+
+ repo = Path(__file__).resolve().parents[2]
+ whl = repo / "android" / "vendor" / "lxst-0.4.8-py3-none-any.whl"
+ assert whl.is_file()
+ with zipfile.ZipFile(whl) as zin:
+ telephony = zin.read("LXST/Primitives/Telephony.py").decode()
+ codecs_init = zin.read("LXST/Codecs/__init__.py").decode()
+ assert "if Codec2 is not None:" in telephony
+ assert "_CODEC2_IMPORT_ERROR" in codecs_init
+
+
def test_repack_script_bundles_libcodec2(tmp_path):
import importlib.util
import zipfile
diff --git a/tests/backend/test_config_manager.py b/tests/backend/test_config_manager.py
index a5e39805..ad112fe2 100644
--- a/tests/backend/test_config_manager.py
+++ b/tests/backend/test_config_manager.py
@@ -84,6 +84,16 @@ def test_telephony_config(db):
config.telephone_allow_calls_from_contacts_only.set(False)
assert config.telephone_allow_calls_from_contacts_only.get() is False
+ # Default audio profile must match LXST DEFAULT_PROFILE (64)
+ assert config.telephone_audio_profile_id.get() == 64
+ config.telephone_audio_profile_id.set(48)
+ assert config.telephone_audio_profile_id.get() == 48
+
+ # Legacy invalid profile 2 is migrated to 64 on reload
+ config.telephone_audio_profile_id.set(2)
+ config3 = ConfigManager(db)
+ assert config3.telephone_audio_profile_id.get() == 64
+
# Test Call Recording
assert config.call_recording_enabled.get() is False
config.call_recording_enabled.set(True)
diff --git a/tests/backend/test_lxst_integration.py b/tests/backend/test_lxst_integration.py
index 1306d318..08c448bf 100644
--- a/tests/backend/test_lxst_integration.py
+++ b/tests/backend/test_lxst_integration.py
@@ -148,10 +148,14 @@ def test_lxst_switch_profile_updates_codec_and_frame_time(monkeypatch):
profile=LXSTTelephony.Profiles.QUALITY_MEDIUM,
filters=[],
packetizer=MagicMock(),
+ audio_source=MagicMock(),
)
telephone.transmit_mixer = _FakeMixer(target_frame_ms=60, gain=0.0)
+ telephone.receive_mixer = _FakeMixer(target_frame_ms=60, gain=0.0)
+ telephone.receive_mixer.set_source_max_frames = MagicMock()
telephone.audio_input = _FakeLineSource()
telephone.transmit_pipeline = _FakePipeline()
+ telephone.target_buffer_frames = 2
telephone.switch_profile(LXSTTelephony.Profiles.QUALITY_HIGH, from_signalling=True)
diff --git a/tests/backend/test_telephone_audio_profiles.py b/tests/backend/test_telephone_audio_profiles.py
new file mode 100644
index 00000000..e858fd04
--- /dev/null
+++ b/tests/backend/test_telephone_audio_profiles.py
@@ -0,0 +1,136 @@
+# SPDX-License-Identifier: 0BSD
+
+"""Telephone audio profile selection and Codec2 readiness."""
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from meshchatx.src.backend.telephone_manager import TelephoneManager
+
+pytest.importorskip("LXST")
+from LXST.Primitives.Telephony import Profiles
+
+
+@pytest.fixture
+def tm():
+ manager = TelephoneManager(identity=MagicMock())
+ manager.telephone = MagicMock()
+ manager.telephone.busy = False
+ manager.telephone.call_status = 3
+ manager.telephone.active_call = None
+ manager._path_poll_interval_s = 0.005
+ manager._path_retry_interval_s = 0.01
+ manager._status_poll_interval_s = 0.01
+ return manager
+
+
+def test_resolve_invalid_profile_falls_back_to_default(tm):
+ assert tm.resolve_audio_profile_id(2) == Profiles.DEFAULT_PROFILE
+ assert tm.resolve_audio_profile_id(999) == Profiles.DEFAULT_PROFILE
+
+
+def test_resolve_codec2_profile_when_available(tm):
+ with patch.object(TelephoneManager, "codec2_available", return_value=True):
+ assert (
+ tm.resolve_audio_profile_id(Profiles.BANDWIDTH_LOW)
+ == Profiles.BANDWIDTH_LOW
+ )
+
+
+def test_resolve_codec2_profile_falls_back_when_unavailable(tm):
+ with patch.object(TelephoneManager, "codec2_available", return_value=False):
+ assert (
+ tm.resolve_audio_profile_id(Profiles.BANDWIDTH_ULTRA_LOW)
+ == Profiles.DEFAULT_PROFILE
+ )
+
+
+def test_apply_preferred_profile_stores_without_idle_switch(tm):
+ tm.telephone.call_status = 3
+ tm.telephone.active_call = None
+ resolved = tm.apply_preferred_profile(Profiles.QUALITY_HIGH)
+ assert resolved == Profiles.QUALITY_HIGH
+ assert tm.preferred_profile_id == Profiles.QUALITY_HIGH
+ tm.telephone.switch_profile.assert_not_called()
+
+
+def test_apply_preferred_profile_switches_when_established(tm):
+ tm.telephone.call_status = 6
+ tm.telephone.active_call = MagicMock()
+ tm.apply_preferred_profile(Profiles.QUALITY_HIGH)
+ tm.telephone.switch_profile.assert_called_once_with(Profiles.QUALITY_HIGH)
+
+
+@pytest.mark.asyncio
+async def test_initiate_passes_preferred_profile_to_lxst_call(tm):
+ destination_hash = bytes.fromhex("aa" * 16)
+ tm.preferred_profile_id = Profiles.BANDWIDTH_LOW
+ seen = {}
+
+ def capture_call(identity, profile=None):
+ seen["identity"] = identity
+ seen["profile"] = profile
+ tm.telephone.call_status = 0
+
+ tm.telephone.call.side_effect = capture_call
+
+ with (
+ patch(
+ "meshchatx.src.backend.telephone_manager.RNS.Identity.recall",
+ return_value=MagicMock(),
+ ),
+ patch(
+ "meshchatx.src.backend.telephone_manager.RNS.Transport.has_path",
+ return_value=True,
+ ),
+ patch.object(TelephoneManager, "codec2_available", return_value=True),
+ patch(
+ "meshchatx.src.backend.telephone_manager.RNS.Destination",
+ ) as dest_cls,
+ ):
+ dest_cls.return_value.hash = destination_hash
+ await tm.initiate(destination_hash, timeout_seconds=1)
+
+ assert seen["profile"] == Profiles.BANDWIDTH_LOW
+
+
+def test_init_telephone_stores_preferred_profile_not_idle_switch(tmp_path):
+ cfg = MagicMock()
+ cfg.telephone_enabled.get.return_value = True
+ cfg.telephone_audio_profile_id.get.return_value = Profiles.QUALITY_MAX
+
+ with patch(
+ "meshchatx.src.backend.telephone_manager.Telephone",
+ ) as telephone_cls:
+ telephone = telephone_cls.return_value
+ manager = TelephoneManager(
+ identity=MagicMock(),
+ config_manager=cfg,
+ storage_dir=str(tmp_path),
+ )
+ manager.init_telephone()
+ assert manager.preferred_profile_id == Profiles.QUALITY_MAX
+ telephone.switch_profile.assert_not_called()
+
+
+def test_pycodec2_encode_decode_for_lxst_codec2_profiles():
+ """Live validation that Codec2 profiles used on Android calls actually work."""
+ import numpy as np
+ from LXST.Codecs import Codec2
+
+ assert TelephoneManager.codec2_available() is True
+ for pid in (
+ Profiles.BANDWIDTH_ULTRA_LOW,
+ Profiles.BANDWIDTH_VERY_LOW,
+ Profiles.BANDWIDTH_LOW,
+ ):
+ codec = Profiles.get_codec(pid)
+ assert isinstance(codec, Codec2)
+ spf = codec.c2.samples_per_frame()
+ pcm = (0.1 * np.sin(np.linspace(0, 8 * np.pi, spf))).astype(np.float32)
+ pcm_i16 = (pcm * 32767).astype(np.int16)
+ encoded = codec.c2.encode(pcm_i16)
+ decoded = codec.c2.decode(encoded)
+ assert len(encoded) == codec.c2.bytes_per_frame()
+ assert len(decoded) == spf
diff --git a/tests/backend/test_telephone_initiation.py b/tests/backend/test_telephone_initiation.py
index 5177a937..e3008d47 100644
--- a/tests/backend/test_telephone_initiation.py
+++ b/tests/backend/test_telephone_initiation.py
@@ -37,7 +37,7 @@ async def test_initiate_retries_path_requests_during_lookup(telephone_manager):
state["calls"] += 1
return state["calls"] >= 8
- telephone_manager.telephone.call.side_effect = lambda _identity: setattr(
+ telephone_manager.telephone.call.side_effect = lambda _identity, *_a, **_k: setattr(
telephone_manager.telephone, "call_status", 0
)
@@ -96,7 +96,7 @@ async def test_initiate_cancels_quickly_while_dialling(telephone_manager):
destination_hash = bytes.fromhex("cc" * 16)
telephone_manager.telephone.call_status = 2
- def blocking_call(_identity):
+ def blocking_call(_identity, *_a, **_k):
time.sleep(0.2)
telephone_manager.telephone.call.side_effect = blocking_call
@@ -162,7 +162,7 @@ async def test_cancel_between_identity_resolved_and_path_request(telephone_manag
async def test_cancel_after_path_found_before_dialling_stabilizes(telephone_manager):
destination_hash = bytes.fromhex("ee" * 16)
- def slow_call(_identity):
+ def slow_call(_identity, *_a, **_k):
time.sleep(0.1)
telephone_manager.telephone.call_status = 0
@@ -208,7 +208,7 @@ async def test_request_path_exceptions_do_not_abort_discovery(telephone_manager)
if isinstance(value, Exception):
raise value
- telephone_manager.telephone.call.side_effect = lambda _identity: setattr(
+ telephone_manager.telephone.call.side_effect = lambda _identity, *_a, **_k: setattr(
telephone_manager.telephone, "call_status", 0
)
@@ -244,7 +244,7 @@ async def test_flapping_path_state_recovers_and_dials(telephone_manager):
return path_states.pop(0)
return True
- telephone_manager.telephone.call.side_effect = lambda _identity: setattr(
+ telephone_manager.telephone.call.side_effect = lambda _identity, *_a, **_k: setattr(
telephone_manager.telephone, "call_status", 0
)
@@ -302,7 +302,7 @@ async def test_call_thread_exception_surfaces_without_hanging(telephone_manager)
async def test_inconsistent_call_status_finishes_within_timeout(telephone_manager):
destination_hash = bytes.fromhex("78" * 16)
- def inconsistent_call(_identity):
+ def inconsistent_call(_identity, *_a, **_k):
telephone_manager.telephone.call_status = 5
telephone_manager.telephone.call.side_effect = inconsistent_call
@@ -335,7 +335,7 @@ async def test_inconsistent_call_status_finishes_within_timeout(telephone_manage
async def test_lxst_status_mapping_updates_ui_initiation_states(telephone_manager):
destination_hash = bytes.fromhex("9a" * 16)
- def status_progression_call(_identity):
+ def status_progression_call(_identity, *_a, **_k):
telephone_manager.telephone.call_status = 2
time.sleep(0.02)
telephone_manager.telephone.call_status = 4
@@ -404,7 +404,7 @@ async def test_rapid_dial_cancel_soak_has_bounded_memory(telephone_manager):
destination_hash = bytes.fromhex("de" * 16)
loops = 120
- def slow_call(_identity):
+ def slow_call(_identity, *_a, **_k):
time.sleep(0.03)
telephone_manager.telephone.call_status = 2
@@ -460,7 +460,7 @@ async def test_initiate_checks_path_for_lxst_telephony_destination(telephone_man
fake_destination = MagicMock()
fake_destination.hash = telephony_destination_hash
- telephone_manager.telephone.call.side_effect = lambda _identity: setattr(
+ telephone_manager.telephone.call.side_effect = lambda _identity, *_a, **_k: setattr(
telephone_manager.telephone, "call_status", 0
)
diff --git a/tests/backend/test_telephone_recorder.py b/tests/backend/test_telephone_recorder.py
index 710ca1e0..2b0a4767 100644
--- a/tests/backend/test_telephone_recorder.py
+++ b/tests/backend/test_telephone_recorder.py
@@ -121,7 +121,7 @@ def test_audio_profile_persistence(mock_identity, mock_config, temp_storage):
"meshchatx.src.backend.telephone_manager.Telephone",
) as mock_telephone_class:
mock_telephone = mock_telephone_class.return_value
- mock_config.telephone_audio_profile_id.get.return_value = 4
+ mock_config.telephone_audio_profile_id.get.return_value = 80
tm = TelephoneManager(
mock_identity,
@@ -130,8 +130,9 @@ def test_audio_profile_persistence(mock_identity, mock_config, temp_storage):
)
tm.init_telephone()
- # Verify switch_profile was called with configured ID
- mock_telephone.switch_profile.assert_called_with(4)
+ # Verify preferred profile is stored for outbound call() (switch_profile is idle no-op)
+ assert tm.preferred_profile_id == 80
+ mock_telephone.switch_profile.assert_not_called()
@patch("meshchatx.src.backend.telephone_manager.Telephone")
──────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────────